package main

import (
	"context"
	"log"

	"github.com/segmentio/kafka-go"
)

func main() {
	// make a writer that produces to topic-A, using the least-bytes distribution
	w := &kafka.Writer{
		Addr:  kafka.TCP("10.122.202.200:9092"),
		Topic: "biz_supplier_rank_detail",
		//Balancer: &kafka.LeastBytes{},
		//Balancer: &kafka.RoundRobin{},
		Balancer: &kafka.Hash{}, //该策略可以保证发送消息时，会按照算法：hasher.Sum32() % len(partitions) => partition，计算得到分区id，所以相同的key会发送到同样的分区;如果生产时不传key，即使设置策略为hash，也是会平均发到n个partition
	}

	msg := `{"supplier_sku_id":44801036,"supplier_id":859222,"index_id":171803131,"detail":"{\"orderCompleteRate\":0.0001,\"commissionRate\":0.0001,\"goClickRate\":0.0001,\"saleNum\":1,\"performanceRate\":0.0001,\"goPv\":1,\"nearly30DaysHits\":50,\"nearly7DaysSaleNum\":0}"}`
	err := w.WriteMessages(context.Background(),
		kafka.Message{
			//Key:   []byte("Key-A"),
			Value: []byte(msg),
		},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	if err := w.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}
}
